package com.zjw.zy.service.helper;

import com.zjw.zy.base.publisher.DefaultPublisher;
import com.zjw.zy.base.publisher.IPublisher;
import com.zjw.zy.bean.User;
import com.zjw.zy.bean.WsMessage;
import com.zjw.zy.constant.AppConstant;
import com.zjw.zy.pal.WsMessageWrap;
import com.zjw.zy.pal.WsStreamProcessor;
import com.zjw.zy.service.contract.IWbSocketService;
import com.zjw.zy.smi.JustWithIt;
import com.zjw.zy.utils.AppUtils;
import com.zjw.zy.utils.GsonUtils;
import com.zjw.zy.utils.LogUtils;
import com.zjw.zy.utils.taskpool.TaskPool;

import org.java_websocket.WebSocket;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft;
import org.java_websocket.framing.Framedata;
import org.java_websocket.handshake.ServerHandshake;

import java.net.URI;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;
import remotecall.call.IRemoteCall;

/**
 * @author ：zhong.jw
 * @date ：Created in 2022/12/4 21:21
 */
public class WbSocketHelper implements IWbSocketService {

    private static final String TAG = "WechatHelper";

    private WechatSocketClient client;

    private volatile boolean markOpen;

    private final IPublisher<IRemoteCall<WsMessage>> mListener = new DefaultPublisher<>();


    private final Map<Long, List<WsMessageWrap.StreamItem>> streamMap = new ConcurrentHashMap<>();

    private static class SingleHolder {
        public static WbSocketHelper INSTANCE = new WbSocketHelper();
    }

    public static WbSocketHelper getInstance() {
        return SingleHolder.INSTANCE;
    }


    public void registerMessageObserver(IRemoteCall<WsMessage> call) {
        mListener.subscribe(call);
    }

    public void unregisterMessageObserver(IRemoteCall<WsMessage> call) {
        mListener.unsubscribe(call);
    }

    private static Disposable sDisposable;

    @Override
    public void connectService(JustWithIt<Boolean> call) {
        TaskPool.concurrent().post(() -> {
            LogUtils.d(TAG, "start connect wechat service");
            markOpen = true;
            if (client != null && client.isOpen()) {
                LogUtils.d(TAG, "connectService isOpen");
                call.onWith(true);
                return;
            }
            if (client != null) {
                repeatConnect();
                return;
            }
            User self = AppUtils.getSelf();
            if (self == null) {
                LogUtils.d(TAG, "AppUtils.getSelf() == null!!!");
                return;
            }
            if (client == null) {
                URI serverUri = URI.create(AppConstant.WECHAT_WS_URL + self.getUsername());
                client = new WechatSocketClient(serverUri);
            }
            try {
                boolean connect = client.connectBlocking();
                if (connect) {
                    startPingPong();
                }
                call.onWith(connect);
            } catch (InterruptedException e) {
                LogUtils.e(TAG, e.getMessage());
                call.onWith(false);
            }
        });
    }

    @Override
    public void disconnectService(JustWithIt<Boolean> call) {
        TaskPool.concurrent().post(() -> {
            markOpen = false;
            stopPingPong();
            if (client != null && client.isOpen()) {
                client.close();
                client = null;
                call.onWith(true);
            }
        });
    }

    @Override
    public void sendMessage(WsMessage message) {
        TaskPool.concurrent().post(() -> {
            if (client != null && client.isOpen()) {
                if (WsStreamProcessor.sendByStream(message, obj -> client.send(ByteBuffer.wrap(obj)))) {
                    return;
                }
                try {
                    client.send(GsonUtils.toJson(message));
                } catch (Exception ignore) {
                }
            } else {
                //否则重连
                connectService(obj -> {
                    if (obj) {
                        sendMessage(message);
                    }
                });
            }
        });
    }


    private void repeatConnect() {
        if (markOpen) {
            stopPingPong();
            TaskPool.concurrent().postDelayed(() -> {
                if (client != null) {
                    LogUtils.d(TAG, "尝试重连wechat");
                    boolean reconnect = false;
                    try {
                        reconnect = client.reconnectBlocking();
                        if (reconnect) {
                            startPingPong();
                        }
                        LogUtils.d(TAG, "重连结果 " + reconnect);
                    } catch (Exception e) {
                        LogUtils.e(TAG, "repeatConnect error:" + e.getMessage());
                    }

                    if (!reconnect) {
                        repeatConnect();
                    }
                }
            }, 10_000L);
        }
    }

    private void startPingPong() {
        stopPingPong();

        LogUtils.d(TAG, "start ping");
        sDisposable = Observable.interval(0, 10, TimeUnit.SECONDS).subscribeOn(
                Schedulers.computation()
        ).subscribe(aLong -> {
            try {
                client.sendPing();
                //服务器心跳，保持连接
                LogUtils.d(TAG, "ping");
            } catch (Exception e) {
                LogUtils.d(TAG, "ping failed");
            }
        });
    }


    private void stopPingPong() {
        LogUtils.d(TAG, "stop ping");
        if (sDisposable != null && !sDisposable.isDisposed()) {
            sDisposable.dispose();
        }
    }


    private class WechatSocketClient extends WebSocketClient {

        public WechatSocketClient(URI serverUri) {
            super(serverUri);
        }

        public WechatSocketClient(URI serverUri, Draft protocolDraft) {
            super(serverUri, protocolDraft);
        }

        public WechatSocketClient(URI serverUri, Map<String, String> httpHeaders) {
            super(serverUri, httpHeaders);
        }

        public WechatSocketClient(URI serverUri, Draft protocolDraft, Map<String, String> httpHeaders) {
            super(serverUri, protocolDraft, httpHeaders);
        }

        public WechatSocketClient(URI serverUri, Draft protocolDraft, Map<String, String> httpHeaders, int connectTimeout) {
            super(serverUri, protocolDraft, httpHeaders, connectTimeout);
        }

        @Override
        public void onOpen(ServerHandshake handshakedata) {
            LogUtils.d(TAG, "onOpen");
        }

        @Override
        public void onMessage(String message) {
            LogUtils.d(TAG, "receive message" + message);
            WsMessage wsMessage = GsonUtils.fromJson(message, WsMessage.class);
            mListener.publish(obj -> obj.call(wsMessage));
        }

        @Override
        public void onMessage(ByteBuffer bytes) {
            //此处的数据是分段数据，分段接收
            WsStreamProcessor.acceptBytes(bytes.array(), message -> {
                mListener.publish(obj -> obj.call(message));
            });
        }

        @Override
        public void onWebsocketPong(WebSocket conn, Framedata f) {
            LogUtils.d(TAG, "pong");
            super.onWebsocketPong(conn, f);
        }

        @Override
        public void onClose(int code, String reason, boolean remote) {
            LogUtils.d(TAG, "onClose " + reason);
            repeatConnect();
        }

        @Override
        public void onError(Exception ex) {
            LogUtils.e(TAG, "onError:" + ex.getMessage());
            //repeatConnect();
        }
    }

}
